package org.jetlinks.pro.messaging.kafka;

import lombok.Generated;
import org.jetlinks.pro.datasource.Command;
import org.jetlinks.pro.datasource.DataSource;
import org.jetlinks.pro.datasource.DataSourceType;
import org.jetlinks.pro.messaging.kafka.command.CreateConsumer;
import org.jetlinks.pro.messaging.kafka.command.CreateProducer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import reactor.core.publisher.Mono;

import javax.annotation.Nonnull;
import java.util.List;

@Generated
public interface KafkaDataSource extends DataSource {

    DataSourceType type = Type.instance;

    KafkaProperties getConfig();

    KafkaProperties copyConfig();

    @Nonnull
    @Override
    <R> R execute(@Nonnull Command<R> command);

    default Mono<KafkaProducer> createProducer() {
        return createProducer(getConfig());
    }

    default Mono<KafkaProducer> createProducer(KafkaProperties properties) {
        return execute(new CreateProducer(properties));
    }

    default Mono<KafkaConsumer> createConsumer(List<String> topics) {
        return execute(new CreateConsumer(topics, getConfig()));
    }

    default Mono<KafkaConsumer> createConsumer(List<String> topics,
                                               KafkaProperties properties) {
        return execute(new CreateConsumer(topics, properties));
    }

    @Generated
    enum Type implements DataSourceType {
        instance;

        @Override
        public String getId() {
            return "kafka";
        }

        @Override
        public String getName() {
            return "Kafka";
        }
    }
}
